package com.mfs.proxy.core.client.reactor;

import com.alibaba.fastjson2.JSON;
import com.mfs.common.constant.Constant;
import com.mfs.common.utills.ChannelUtil;
import com.mfs.common.utills.ObjectUtil;
import com.mfs.proxy.core.client.holder.ProxyServerHolder;
import com.mfs.proxy.core.exceptions.ProxyException;
import com.mfs.proxy.core.client.config.ClientConfig;
import com.mfs.proxy.core.server.endpoint.ProxyHandle;
import com.mfs.proxy.core.util.ProxyTaskExecutor;
import com.mfs.proxy.core.enums.MessageTypeEnum;
import com.mfs.proxy.core.message.mailbox.CommunicationMailbox;
import com.mfs.proxy.core.message.mailbox.DefaultCommunicationMailbox;
import com.mfs.proxy.core.message.Message;
import com.mfs.proxy.core.server.ProxyServer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;

/**
 * 客户端reactor，建立与代理服务端的消息连接，主动发送心跳消息，接收服务端通道请求
 *
 * @author mfs
 * @date 2023-07-04
 * */
public class ClientReactor implements Runnable {
    private static final Logger logger = LoggerFactory.getLogger(ClientReactor.class);
    private Selector clientSelector;
    public ClientReactor() {
        try {
            clientSelector = Selector.open();
            SocketChannel serverChannel = SocketChannel.open(new InetSocketAddress(ClientConfig.getInstance().getServerIp(),
                    ClientConfig.getInstance().getServerPort()));
            serverChannel.configureBlocking(false);
            serverChannel.register(clientSelector, SelectionKey.OP_WRITE);
        } catch (IOException e) {
            throw new ProxyException(e);
        }
    }

    @Override
    public void run() {
        try {
            while (!Thread.currentThread().isInterrupted() && clientSelector.keys().size() > 0) {
                clientSelector.select();
                Iterator<SelectionKey> iterator = clientSelector.selectedKeys().iterator();
                while(iterator.hasNext()) {
                    processReadableWritable(iterator.next());
                    iterator.remove();
                }
            }
        } catch (IOException e) {
            throw new ProxyException(e);
        }
    }

    /**
     * 处理首次与代理服务端连接时的writable，发送{@link com.mfs.proxy.core.enums.MessageTypeEnum#CONNECT}消息
     * 发送完消息后立刻转为OP_READALE, 等待接收服务端的消息
     * */
    private void processReadableWritable(SelectionKey selectionKey) {
        if (selectionKey.isValid() && selectionKey.isWritable()) {
            CommunicationMailbox communicationMailbox = new DefaultCommunicationMailbox(selectionKey);
            try {
                communicationMailbox.send(MessageTypeEnum.CONNECT);
            } catch (IOException e) {
                throw new ProxyException(e);
            }
            ProxyServerHolder.setProxyServer(new ProxyServer(communicationMailbox, ProxyTaskExecutor.getInstance()));
        } else if (selectionKey.isValid() && selectionKey.isReadable()) {
            // 1. 获取message对象
            SocketChannel serverChannel = (SocketChannel) selectionKey.channel();
            byte[] serverMsg = new byte[0];
            try {
                serverMsg = ChannelUtil.read(serverChannel);
            } catch (IOException e) {
                throw new ProxyException("读取信息失败，服务端主动断开连接", e);
            }
            if (serverMsg.length < Constant.ONE) {
                // 客户端断开连接
                try {
                    serverChannel.close();
                } catch (IOException e) {
                    logger.error("连接已关闭", e);
                }
                throw new ProxyException("服务端主动断开连接");
            }
            Message message = ObjectUtil.deserialize(serverMsg, Message.class);

            // 2. 匹配消息枚举
            MessageTypeEnum messageEnum = MessageTypeEnum.valueOf(message.getMessageType());
            if (messageEnum == null) {
                logger.error("不能处理的消息类型:{}", JSON.toJSONString(message));
                return;
            }

            // 3.消息处理
            Object messageContent = message.getMessageContent();
            logger.info("收到客户端消息,消息类型：{}，消息内容：{}", messageEnum.getDescribe(), messageContent);
            if (messageEnum == MessageTypeEnum.CONNECTED) {
                ProxyHandle proxyHandle = (ProxyHandle) messageContent;
                logger.info("接入服务端成功，代理地址 {}:{}", proxyHandle.getProxyIp(), proxyHandle.getProxyPort());
            }
            try {
                messageEnum.getMessageReceiveProcessor().receive(selectionKey, messageContent);
            } catch (IOException e) {
                logger.error("处理客户端消息异常", e);
                try {
                    selectionKey.channel().close();
                } catch (IOException ioException) {
                    ioException.printStackTrace();
                }
                selectionKey.cancel();
                throw new ProxyException("处理客户端消息异常", e);
            }
        }
    }
}
